package com.example.springbootesservice.service;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.example.springbootesservice.common.*;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.query.*;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.*;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedLongTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.*;
import org.elasticsearch.search.aggregations.pipeline.BucketSelectorPipelineAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @author wuxinxin
 */
@Service
@Slf4j
public class EsApiService {

    @Autowired
    private RestHighLevelClient restHighLevelClient;


    public void test() throws IOException {

        SearchRequest searchRequest = new SearchRequest("myindex_new");
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.from(0);
        searchSourceBuilder.size(10);

        searchRequest.source(searchSourceBuilder);

        SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        SearchHits hits = search.getHits();

        for (SearchHit searchHit:hits){
            System.out.println(JSONUtil.toJsonStr(searchHit.getSourceAsString()));
        }

    }

    /**
     * 创建索引
     * @param createIndexDto
     * @throws IOException
     */
    public void createIndex(CreateIndexDto createIndexDto) throws IOException {
        CreateIndexRequest indexRequest = new CreateIndexRequest(createIndexDto.getIndexName());

        indexRequest.settings(Settings.builder()
                .put("number_of_shards",createIndexDto.getShards())
                .put("number_of_replicas",createIndexDto.getReplicas())
        );

        //创建索引
        restHighLevelClient.indices().create(indexRequest, RequestOptions.DEFAULT);
    }

    /**
     * 添加文档
     * @param addDocDto
     * @throws IOException
     */
    public void addDoc(AddDocDto addDocDto) throws IOException {

        IndexRequest indexRequest = new IndexRequest(addDocDto.getIndexName());

        indexRequest.source(addDocDto.getData());
        indexRequest.id(addDocDto.getId());

        log.info(indexRequest.toString());
        //创建索引
        IndexResponse index = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);

        log.info("执行结果:"+JSONUtil.toJsonStr(index));
    }

    /**
     * 更新文档(根据id插入文档，发现存在会触发更新操作，是增量字段)
     * @param updateDocDto
     * @throws IOException
     */
    public void updateDoc(UpdateDocDto updateDocDto) throws IOException {

        IndexRequest indexRequest = new IndexRequest(updateDocDto.getIndexName());

        indexRequest.source(updateDocDto.getData());
        indexRequest.id(updateDocDto.getId());

        log.info(indexRequest.toString());
        //创建索引
        IndexResponse index = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);

        log.info("执行结果:"+JSONUtil.toJsonStr(index));
    }

    /**
     * 更新文档（根据条件更新数据，删掉了老字段）
     * @param updateDocDto
     * @throws IOException
     */
    public void updateDoc2(UpdateDocDto updateDocDto) throws IOException {

        UpdateByQueryRequest indexRequest = new UpdateByQueryRequest(updateDocDto.getIndexName());

        //设置更新条件
        Set<String> strings = updateDocDto.getQuery().keySet();
        for (String key:strings){
            QueryBuilder queryBuilder = new TermQueryBuilder(key,updateDocDto.getQuery().get(key));
            indexRequest.setQuery(queryBuilder);
            log.info(queryBuilder.toString());
        }
        //设置更新值
        JSONObject data = updateDocDto.getData();

        //更新值脚本
        String source = "ctx._source=params";
        Script script = new Script(ScriptType.INLINE, "painless", source, data);
        indexRequest.setScript(script);

        log.info(indexRequest.toString());
        //更新文档
        BulkByScrollResponse index = restHighLevelClient.updateByQuery(indexRequest, RequestOptions.DEFAULT);

        log.info("执行结果:"+JSONUtil.toJsonStr(index));
    }

    /**
     * 更新文档(仅仅根据id更新，是增量字段)
     * @param updateDocDto
     * @throws IOException
     */
    public void updateDoc3(UpdateDocDto updateDocDto) throws IOException {

        UpdateRequest indexRequest = new UpdateRequest(updateDocDto.getIndexName(),updateDocDto.getId());

        //设置数据
        indexRequest.doc(updateDocDto.getData());

        //更新文档
        UpdateResponse update = restHighLevelClient.update(indexRequest, RequestOptions.DEFAULT);

        log.info("执行结果:"+JSONUtil.toJsonStr(update));
    }


    /**
     * 删除文档（根据id）
     * @param deleteDocDto
     * @throws IOException
     */
    public void deleteDoc(DeleteDocDto deleteDocDto) throws IOException {

        DeleteRequest indexRequest = new DeleteRequest(deleteDocDto.getIndexName(),deleteDocDto.getId());

        //更新文档
        DeleteResponse update = restHighLevelClient.delete(indexRequest, RequestOptions.DEFAULT);

        log.info("执行结果:"+JSONUtil.toJsonStr(update));
    }

    /**
     * 删除文档(根据任意条件)
     * @param deleteDocDto
     * @throws IOException
     */
    public void deleteDoc2(DeleteDocDto deleteDocDto) throws IOException {

        DeleteByQueryRequest indexRequest = new DeleteByQueryRequest(deleteDocDto.getIndexName());

        //设置查询条件
        Set<String> strings = deleteDocDto.getQuery().keySet();
        for (String key:strings){
            QueryBuilder queryBuilder = new TermQueryBuilder(key,deleteDocDto.getQuery().get(key));
            indexRequest.setQuery(queryBuilder);
            log.info(queryBuilder.toString());
        }
        //更新文档
        BulkByScrollResponse response = restHighLevelClient.deleteByQuery(indexRequest, RequestOptions.DEFAULT);

        log.info("执行结果:"+JSONUtil.toJsonStr(response));
    }


    /**
     * 单字段查询(等值)
     * @param searchDocDto
     * @throws IOException
     */
    public void search(SearchDocDto searchDocDto) throws IOException {

        //构件查询
        TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(searchDocDto.getKey(), searchDocDto.getValue());
        //查询
        searchCommon(searchDocDto.getIndexName(),termQueryBuilder);
    }

    /**
     * 单字段查询(多值等值)
     * @param searchDocDto
     * @throws IOException
     */
    public void search2(SearchDocDto searchDocDto) throws IOException {

        //构件查询
        TermsQueryBuilder termsQueryBuilder = QueryBuilders.termsQuery(searchDocDto.getKey(), searchDocDto.getValues());
        //查询
        searchCommon(searchDocDto.getIndexName(),termsQueryBuilder);
    }

    /**
     * 单字段查询(范围)
     * @param searchDocDto
     * @throws IOException
     */
    public void search3(SearchDocDto searchDocDto) throws IOException {
        //构件查询
        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(searchDocDto.getKey());
        rangeQueryBuilder.from(searchDocDto.getStartValue());
        rangeQueryBuilder.to(searchDocDto.getEndValue());

        //查询
        searchCommon(searchDocDto.getIndexName(),rangeQueryBuilder);
    }

    /**
     * 复合多字段查询(should)
     * @param searchDocDto
     */
    public void search4(SearchDocDto searchDocDto) throws IOException {
        //构件查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        JSONObject query = searchDocDto.getQuery();
        Set<String> strings = query.keySet();

        for(String key:strings) {
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(key, query.get(key));
            boolQueryBuilder.should(termQueryBuilder);
        }
        //查询
        searchCommon(searchDocDto.getIndexName(),boolQueryBuilder);
    }


    /**
     * 复合多字段查询(must)
     * @param searchDocDto
     */
    public void search5(SearchDocDto searchDocDto) throws IOException {
        //构件查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        JSONObject query = searchDocDto.getQuery();
        Set<String> strings = query.keySet();

        for(String key:strings) {
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(key, query.get(key));
            boolQueryBuilder.must(termQueryBuilder);
        }

        //查询
        searchCommon(searchDocDto.getIndexName(),boolQueryBuilder);
    }

    /**
     * 复合多字段查询(must not,不计算相关性分数，和must查询到数据相反)
     * @param searchDocDto
     */
    public void search6(SearchDocDto searchDocDto) throws IOException {
        //构件查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        JSONObject query = searchDocDto.getQuery();
        Set<String> strings = query.keySet();

        for(String key:strings) {
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(key, query.get(key));
            boolQueryBuilder.mustNot(termQueryBuilder);
        }
        //查询
        searchCommon(searchDocDto.getIndexName(),boolQueryBuilder);
    }

    /**
     * 复合多字段查询(filter,不计算相关性分数，和must查询到数据相同)
     * @param searchDocDto
     */
    public void search7(SearchDocDto searchDocDto) throws IOException {
        //构件查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        JSONObject query = searchDocDto.getQuery();
        Set<String> strings = query.keySet();

        for(String key:strings) {
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(key, query.get(key));
            boolQueryBuilder.filter(termQueryBuilder);
        }
        //查询
        searchCommon(searchDocDto.getIndexName(),boolQueryBuilder);
    }


    /**
     * 复合多字段查询(should,must,must_not,filter进行复合)
     * @param searchDocDto
     */
    public void search8(SearchDocDto searchDocDto) throws IOException {
        //构件查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        //构件should条件
        JSONObject query = searchDocDto.getShouldQuery();
        Set<String> strings = query.keySet();

        for(String key:strings) {
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(key, query.get(key));
            boolQueryBuilder.should(termQueryBuilder);
        }
        //构件must条件
        JSONObject query1 = searchDocDto.getMustQuery();
        Set<String> strings1 = query1.keySet();

        for(String key:strings1) {
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(key, query1.get(key));
            boolQueryBuilder.must(termQueryBuilder);
        }
        //查询
        searchCommon(searchDocDto.getIndexName(),boolQueryBuilder);
    }

    /**
     * 复合多字段查询(（should,must,must_not,filter）（bool）进行复合)
     * @param searchDocDto
     */
    public void search9(SearchDocDto searchDocDto) throws IOException {
        //构件查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        //构件should条件
        JSONObject query = searchDocDto.getShouldQuery();
        Set<String> strings = query.keySet();

        for(String key:strings) {
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(key, query.get(key));
            boolQueryBuilder.should(termQueryBuilder);
        }
        //构件内部bool
        BoolQueryBuilder boolQueryBuilder1 = QueryBuilders.boolQuery();
        JSONObject query1 = searchDocDto.getInnerBoolQuery();
        Set<String> strings1 = query1.keySet();
        for(String key:strings1) {
            TermQueryBuilder termQueryBuilder = QueryBuilders.termQuery(key, query1.get(key));
            boolQueryBuilder1.must(termQueryBuilder);
        }

        boolQueryBuilder.should(boolQueryBuilder1);
        //查询
        searchCommon(searchDocDto.getIndexName(),boolQueryBuilder);
    }

    /**
     * 复合多字段查询(（should,must,must_not,filter）（term,much,range）进行复合)
     * @param searchDocDto
     */
    public void search10(SearchDocDto searchDocDto) throws IOException {
        //构件range条件
        RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(searchDocDto.getKey());
        rangeQueryBuilder.from(searchDocDto.getStartValue());
        rangeQueryBuilder.to(searchDocDto.getEndValue());

        //构件should条件
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.should(rangeQueryBuilder);

        //查询
        searchCommon(searchDocDto.getIndexName(),boolQueryBuilder);
    }

    /**
     * 单字段查询（模糊-分词查询）
     * @param searchDocDto
     */
    public void search11(SearchDocDto searchDocDto) throws IOException {
        //构件match查询
        MatchQueryBuilder matchQueryBuilder = QueryBuilders.matchQuery(searchDocDto.getKey(), searchDocDto.getValue());

        //查询
        searchCommon(searchDocDto.getIndexName(),matchQueryBuilder);
    }

    /**
     * 多字段查询（模糊多个字段-分词查询）
     * @param searchDocDto
     */
    public void search12(SearchDocDto searchDocDto) throws IOException {
        //构件match查询
        MultiMatchQueryBuilder multiMatchQueryBuilder = QueryBuilders.multiMatchQuery(searchDocDto.getValue(),searchDocDto.getFields());

        //查询
        searchCommon(searchDocDto.getIndexName(),multiMatchQueryBuilder);
    }

    /**
     * 单字段查询（模糊-词组分词查询）
     * @param searchDocDto
     */
    public void search13(SearchDocDto searchDocDto) throws IOException {
        //构件match查询
        MatchPhraseQueryBuilder matchPhraseQueryBuilder = QueryBuilders.matchPhraseQuery(searchDocDto.getKey(), searchDocDto.getValue());

        //查询
        searchCommon(searchDocDto.getIndexName(),matchPhraseQueryBuilder);
    }


    /**
     * 分页
     * @param searchDocDto
     */
    public void search14(SearchDocDto searchDocDto) throws IOException {

        SearchRequest searchRequest = new SearchRequest(searchDocDto.getIndexName());
        //构建查询
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //分页
        searchSourceBuilder.from(1);
        searchSourceBuilder.size(2);

        searchRequest.source(searchSourceBuilder);

        log.info("查询条件"+searchSourceBuilder.toString());

        SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        SearchHits hits = search.getHits();

        for (SearchHit searchHit:hits){
            log.info(JSONUtil.toJsonStr(searchHit.getSourceAsString()));
        }
    }

    /**
     * 聚合(terms--统计文档数量)==(group(可以嵌套group,相当于多字段group)+count(*)聚合函数)
     * 聚合（max）==(max(*)聚合函数)
     * 聚合（min）==(min(*)聚合函数)
     * 聚合（avg）==(avg(*)聚合函数)
     * 聚合（sum）==(sum(*)聚合函数)
     * 聚合（top_hits）==保存聚合后记录
     * @param searchDocDto
     * @throws IOException
     */
    public void search15(SearchDocDto searchDocDto) throws IOException {

        SearchRequest searchRequest = new SearchRequest(searchDocDto.getIndexName());
        //构建查询
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //聚合
        TermsAggregationBuilder ageAgg = AggregationBuilders.terms("ageAgg");
        //设置聚合字段
        ageAgg.field(searchDocDto.getAggField());
        //设置展示聚合分布的结果数量
        ageAgg.size(searchDocDto.getAggSize());
        //设置聚合结果文档数下限
        ageAgg.minDocCount(searchDocDto.getAggMinDocCount());
        //设置排序
        BucketOrder count = BucketOrder.count(true);
        ageAgg.order(count);

        //设置子聚合
        AggregatorFactories.Builder builder = AggregatorFactories.builder();
        //设置平均值聚合函数
        AvgAggregationBuilder balanceAvgAgg = AggregationBuilders.avg("balanceAvgAgg");
        balanceAvgAgg.field("balance2");
        builder.addAggregator(balanceAvgAgg);

        //设置最大值聚合函数
        MaxAggregationBuilder maxAgg = AggregationBuilders.max("balanceMaxAgg");
        maxAgg.field("balance2");
        builder.addAggregator(maxAgg);

        //设置最小值聚合函数
        MinAggregationBuilder minAgg = AggregationBuilders.min("balanceMinAgg");
        minAgg.field("balance2");
        builder.addAggregator(minAgg);

        //设置总数聚合函数
        SumAggregationBuilder sumAggregationBuilder = AggregationBuilders.sum("balanceSumAgg");
        sumAggregationBuilder.field("balance2");
        builder.addAggregator(sumAggregationBuilder);

        //设置聚合中各值对应记录
        TopHitsAggregationBuilder aggRecord = AggregationBuilders.topHits("aggRecord");
        aggRecord.size(1);
        builder.addAggregator(aggRecord);

        ageAgg.subAggregations(builder);

        searchSourceBuilder.aggregation(ageAgg);

        searchRequest.source(searchSourceBuilder);

        log.info("查询条件"+searchSourceBuilder.toString());

        SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        //获取聚合结果
        Aggregations aggregations = search.getAggregations();

        for (Aggregation aggr:aggregations.asList()){
            ParsedLongTerms parsedLongTerms=(ParsedLongTerms)aggr;
            List<? extends Terms.Bucket> buckets = parsedLongTerms.getBuckets();
            for(Terms.Bucket bucket:buckets){
                log.info(JSONUtil.toJsonStr(bucket));
            }
        }
    }


    /**
     * 聚合（count）==(count(distinct*)聚合函数)
     * @param searchDocDto
     * @throws IOException
     */
    public void search16(SearchDocDto searchDocDto) throws IOException {

        SearchRequest searchRequest = new SearchRequest(searchDocDto.getIndexName());
        //构建查询
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //聚合
        CardinalityAggregationBuilder ageAgg = AggregationBuilders.cardinality("ageAgg");
        //设置聚合字段
        ageAgg.field(searchDocDto.getAggField());

        searchSourceBuilder.aggregation(ageAgg);

        searchRequest.source(searchSourceBuilder);

        log.info("查询条件"+searchSourceBuilder.toString());

        SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        //获取聚合结果
        Aggregations aggregations = search.getAggregations();

        for (Aggregation aggr:aggregations.asList()){
            ParsedCardinality parsedLongTerms=(ParsedCardinality)aggr;
            log.info(parsedLongTerms.getValue()+"");
        }
    }

    /**
     * 聚合（bucket_selector）==(having in)聚合函数)
     * @param searchDocDto
     * @throws IOException
     */
    public void search17(SearchDocDto searchDocDto) throws IOException {

        SearchRequest searchRequest = new SearchRequest(searchDocDto.getIndexName());
        //构建查询
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //聚合term聚合
        TermsAggregationBuilder ageAgg = AggregationBuilders.terms("ageAgg");
        ageAgg.field(searchDocDto.getAggField());
        searchSourceBuilder.aggregation(ageAgg);

        //设置子聚合
        AggregatorFactories.Builder builder = AggregatorFactories.builder();
        //设置bucket_selector
        Map<String, String> bucketsPath = new HashMap<>();
        bucketsPath.put("count","_count");
        BucketSelectorPipelineAggregationBuilder pipelineAggregationBuilder = new BucketSelectorPipelineAggregationBuilder
                ("filterAgg",bucketsPath, Script.parse("params.count>1"));
        builder.addPipelineAggregator(pipelineAggregationBuilder);

        ageAgg.subAggregations(builder);

        searchRequest.source(searchSourceBuilder);

        log.info("查询条件"+searchSourceBuilder.toString());

        SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        Aggregations aggregations = search.getAggregations();

        for (Aggregation aggr:aggregations.asList()){
            ParsedLongTerms parsedLongTerms=(ParsedLongTerms)aggr;
            List<? extends Terms.Bucket> buckets = parsedLongTerms.getBuckets();
            for(Terms.Bucket bucket:buckets){
                log.info(JSONUtil.toJsonStr(bucket));
            }
        }
    }

    private void searchCommon(String indexName,QueryBuilder queryBuilder) throws IOException {

        SearchRequest searchRequest = new SearchRequest(indexName);

        //构建查询
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //设置query条件
        searchSourceBuilder.query(queryBuilder);
        searchRequest.source(searchSourceBuilder);

        log.info("查询条件"+searchSourceBuilder.toString());

        SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

        SearchHits hits = search.getHits();

        for (SearchHit searchHit:hits){
            log.info(JSONUtil.toJsonStr(searchHit.getSourceAsString()));
        }
    }
}
